package club.janna.rpcdemo.client.invoke;

import club.janna.rpcdemo.codec.Codec;
import club.janna.rpcdemo.codec.CodecProvider;
import club.janna.rpcdemo.domain.Request;
import club.janna.rpcdemo.domain.Response;
import club.janna.rpcdemo.domain.ServiceInfo;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.concurrent.CountDownLatch;

@Slf4j
public class TCPNetInvoker implements NetInvoker {

    @Autowired
    private CodecProvider codecProvider;

    @Override
    public Response invoke(ServiceInfo serviceInfo, Request request) {
        Codec codec = codecProvider.get(serviceInfo.getProtocol());
        if(codec == null)
            throw new RuntimeException("找不到编解码器，" + serviceInfo.getProtocol());
        Bootstrap bootstrap = new Bootstrap();
        RequestHandler requestHandler = new RequestHandler(request, codec);
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(1);
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        // 报文最大100k，2个字节报文头，4字节报文长度
                        socketChannel.pipeline().addLast(new LengthFieldBasedFrameDecoder(102400, 2, 4, 0, 6));
                        socketChannel.pipeline().addLast(requestHandler);
                    }
                });
        try {
            ChannelFuture future = bootstrap.connect(serviceInfo.getIp(), serviceInfo.getPort()).sync();
            log.debug("连接到服务器 -> [{}, {}]", serviceInfo.getIp(), serviceInfo.getPort());
            future.channel().closeFuture().await();
            eventLoopGroup.shutdownGracefully();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return requestHandler.getResponse();
    }

    private static class RequestHandler extends ChannelInboundHandlerAdapter {
        private Codec codec;
        private Request request;
        private CountDownLatch latch = new CountDownLatch(1);
        private Response response;

        public RequestHandler(Request request, Codec codec) {
            this.codec = codec;
            this.request = request;
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            // 发送请求
            ctx.writeAndFlush(Unpooled.wrappedBuffer(codec.encode(request)));
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            // 获取响应
            ByteBuf buf = (ByteBuf) msg;
            byte[] bytes = new byte[buf.readableBytes()];
            buf.readBytes(bytes);
            buf.release();
            // 解码
            response = codec.decode(bytes);
            latch.countDown();
            ctx.channel().close();
        }

        public Response getResponse() {
            return response;
        }
    }
}
